package com.shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

public class Demo3DianXin {
    private Connection conn;
    private Admin admin;

    @Before
    public void init() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        try {
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    // 将电信数据加载到HBase中
    public void loadDataToHBase() throws IOException {
        // 如果表不存在就创建表
        if (!admin.tableExists(TableName.valueOf("dianxin"))) {
            admin.createTable(new HTableDescriptor(TableName.valueOf("dianxin"))
                    .addFamily(new HColumnDescriptor("info").setMaxVersions(10)));
        }
        BufferedReader bufferedReader = new BufferedReader(new FileReader("data/DIANXIN.csv"));
        String line = bufferedReader.readLine();
        // 获取待插入的表
        Table dianxin = conn.getTable(TableName.valueOf("dianxin"));
        while (line != null) {
            String[] splits = line.split(",");
            String mdn = splits[0];
            String start_time = splits[1];
            // 经度
            String longitude = splits[4];
            // 维度
            String latitude = splits[5];
            String pos = longitude + "," + latitude;
            Put put = new Put(mdn.getBytes());
            // 使用start_time作为时间戳，利用HBase默认保留 ts 最大的10份（建表的时候指定的版本数量）数据
            put.addColumn("info".getBytes(), "pos".getBytes(), Long.parseLong(start_time), pos.getBytes());

            dianxin.put(put);

            // 获取下一行数据
            line = bufferedReader.readLine();
        }

    }


    @Test
    public void getLastPostion() throws IOException {
        Table dianxin = conn.getTable(TableName.valueOf("dianxin"));
        String mdn = "47BE1E866CFC071DB19D5E1C056BE28AE24C16E7";
        Get get = new Get(mdn.getBytes());
        // 设置一下需要获取的版本数量，如果不设置，默认获取最新的一条
        get.setMaxVersions(10);
        Result rs = dianxin.get(get);
        String rowkey = Bytes.toString(rs.getRow());
        System.out.println("rowkey为：" + rowkey);
        for (Cell cell : rs.listCells()) {
            String family = Bytes.toString(CellUtil.cloneFamily(cell));
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
            String position = Bytes.toString(CellUtil.cloneValue(cell));
            System.out.println("family:" + family + ",qualified:" + qualifier + ",position:" + position);
        }

    }


    @After
    public void close() {
        try {
            if (conn != null) {
                conn.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
